package com.raylu.realtime.app.template;

import com.raylu.realtime.bean.MetaBean;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2022/4/8.
 * Copyright © 2022 lucienoz. All rights reserved.
 */
public class MySqlTwoPhaseCommitSink2<IN> extends TwoPhaseCommitSinkFunction<IN, Connection, Void> {


    public MySqlTwoPhaseCommitSink2() {
        super(new KryoSerializer<>(Connection.class, new ExecutionConfig()), VoidSerializer.INSTANCE);
    }

//    public MySqlTwoPhaseCommitSink(TypeSerializer<Connection> transactionSerializer, TypeSerializer<Void> contextSerializer) {
//        super(transactionSerializer, contextSerializer);
//    }


    @Override
    protected void invoke(Connection transaction, IN value, Context context) throws Exception {
        PreparedStatement ps = transaction.prepareStatement("INSERT INTO testdb.meta_data_full ( in_iface, src_ip, src_port, event_type, flow_id, dest_ip, proto, host, http, tx_id, dest_port, timestamp, datatime) " +
                "VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)");
        MetaBean metaBean = (MetaBean) value;
        ps.setString(1, metaBean.getIn_iface());
        ps.setString(2, metaBean.getSrc_ip());
        ps.setInt(3, metaBean.getSrc_port());
        ps.setString(4, metaBean.getEvent_type());
        ps.setString(5, metaBean.getFlow_id());
        ps.setString(6, metaBean.getDest_ip());
        ps.setString(7, metaBean.getProto());
        ps.setString(8, metaBean.getHost());
        ps.setString(9, metaBean.getHttp());
        ps.setString(10, metaBean.getTx_id());
        ps.setInt(11, metaBean.getDest_port());
        ps.setString(12, metaBean.getTimestamp());
        ps.setTimestamp(13, metaBean.getDatatime());

        ps.execute();

        System.out.println("invoke execute");
    }

    @Override
    protected Connection beginTransaction() throws Exception {
        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/testdb?useSSL=false", "root", "root");
        connection.setAutoCommit(false);
        System.out.println("-->"+connection.toString()+"+");
        return connection;
    }

    @Override
    protected void preCommit(Connection transaction) throws Exception {
        System.out.println("preCommit");
//        Long value = commitCount.value();
//        System.out.println("commit count:"+value);
    }

    @Override
    protected void commit(Connection transaction) {
        System.out.println("commit");
        try {
            transaction.commit();
            System.out.println("-->"+transaction.toString()+"-");
            transaction.close();


        } catch (SQLException e) {
            e.printStackTrace();
            System.out.println("commit error");
        }
    }

    @Override
    protected void abort(Connection transaction) {
        try {
            transaction.rollback();
            System.out.println("-->"+transaction.toString()+"-");
            transaction.close();
//            transaction.setAutoCommit(true);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        System.out.println("abort");
    }
}
